package skewreduce.framework.physical;

import java.io.DataInputStream;
import java.io.FileInputStream;
import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.Job;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import skewreduce.framework.EaggDriver;
import skewreduce.framework.EaggPlanCompiler;
import skewreduce.framework.SchedulerEvent;
import skewreduce.framework.logical.LogicalOp;


/**
 * maybe misleading naming since it does not do real optimization.
 * @author yongchul
 *
 */
public class POptimizeOp extends PhysicalOp {
    private static final Logger LOG = LoggerFactory.getLogger("PLAN");

	@Override
	public void setup() throws IOException {}
	
	@Override
	protected Job createJob(Configuration conf) throws IOException {
		throw new UnsupportedOperationException("OPTIMIZE operator does not run a Map-Reduce job");
	}

	@Override
	public SchedulerEvent call() throws Exception {
		// at this point, our bound object should be expanded by the SAMPLE operator

//		String pf = getConf().get("skewreduce.partition.manual.file");
//
//        if ( LOG.isDebugEnabled() ) {
//            LOG.debug("manual parittion file = "+pf);
//        }
//        
//		boolean doDefault = ( pf == null ) || pf.length() == 0;
		
		boolean doDefault = ! plan.hasManualPlan();
		
		if ( ! doDefault ) {
//	      File f = new File(pf);
//            if ( f.exists() ) {
            	// FIXME: do not prune. accept manual plan.
                DataInputStream in = new DataInputStream(new FileInputStream(plan.getManualPartitionFile()));
                getBound().readFields(in);
                in.close();
                
//                EaggPlanCompiler.compileManualPlan(plan, this);
                EaggPlanCompiler.compileManualPlan2(plan, this);
//            } else {
//            	doDefault = true;
//            }
		}
		
		if ( doDefault ) {
			EaggPlanCompiler.compileInitialPlan(plan,this);
		}
      
        setStatus(Status.COMPLETE);
		return new SchedulerEvent.Completion(this);
	}
	
	public static LogicalOp createLocalOp(EaggDriver driver) {
		return new LogicalOp(driver,LogicalOp.Type.OPTIMIZE,0) {
			@Override
			public PhysicalOp createInstance() {
				return new POptimizeOp();
			}
		};
	}
}
